package com.hujf.flink.window;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @Author hujf
 * @Date 2022/5/11 15:41
 * @Description TODO
 */
public class CountTable {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple3<String, Integer, Long>> dataStreamSource = env.addSource(new MySourceFunction());
        DataStream<Tuple3<String, Integer, Long>> result = dataStreamSource.keyBy(0).
                countWindow(4).sum(1);
        result.print();
        env.execute("qingh Demo");
    }

}

class MySourceFunction implements SourceFunction<Tuple3<String, Integer, Long>>{


    @Override
    public void run(SourceContext ctx) throws Exception {
        Tuple3 tuple1 = new Tuple3("1",1,1L);
        ctx.collect(tuple1);

        tuple1 = new Tuple3("2",2,2L);
        ctx.collect(tuple1);
        tuple1 = new Tuple3("2",3,2L);
        ctx.collect(tuple1);
        tuple1 = new Tuple3("2",4,2L);
        ctx.collect(tuple1);
        tuple1 = new Tuple3("2",2,2L);
        ctx.collect(tuple1);
    }

    @Override
    public void cancel() {

    }
}
